package com.econ.powercloud.ems.observer;

import com.econ.powercloud.ems.common.pojo.EmsKafkaYxDataDTO;
import com.econ.powercloud.ems.common.interf.CalStatusDataObserver;
import com.econ.powercloud.ems.websocket.WebsocketObserverManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.user.SimpSession;
import org.springframework.messaging.simp.user.SimpUser;
import org.springframework.messaging.simp.user.SimpUserRegistry;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Set;

/**
 * 计算遥信数据推送
 */
@Component
public class CalStatusDataWebsocketProcessor implements CalStatusDataObserver {

    @Autowired
    WebsocketObserverManager websocketObserverManager;
    @Autowired
    private SimpUserRegistry simpUserRegistry;
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Override
    public void doInvoke(EmsKafkaYxDataDTO data) {
        //一、查找订阅的用户
        Set<SimpUser> users = simpUserRegistry.getUsers();
        for (SimpUser simpUser : users) {
            Set<SimpSession> sessions = simpUser.getSessions();
            for (SimpSession session : sessions) {
                //二、判断是否需要推送
                Set<String> pidSet = websocketObserverManager.getPidSetBySessionId(session.getId());
                if (CollectionUtils.isEmpty(pidSet)) {
                    continue;
                }
                //三、判断是否为订阅的数据
                if(pidSet.contains(data.getPid())) {
                    SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
                    headerAccessor.setSessionId(session.getId());
                    headerAccessor.setLeaveMutable(true);
                    messagingTemplate.convertAndSendToUser(simpUser.getName(), "/topic/device/cal/status", data, headerAccessor.getMessageHeaders());
                }
            }
        }
    }

}
